package com.cn.client;

import com.cn.common.*;
import com.cn.common.consumer.MessageQueueAllLock;
import com.cn.common.processor.ProcessorCode;
import com.cn.exception.MyException;
import com.cn.msg.Message;
import com.cn.remote.InvokeCallback;
import com.cn.remote.RemoteClient2;
import com.cn.util.RemotingSerializable;
import org.apache.commons.lang3.StringUtils;

import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class MqClientApiImpl {

    private RemoteClient2 remoteClient2;
    private NettyClientConfig nettyClientConfig;
    private ConcurrentMap<String /*brokerName*/, TreeMap<Integer, String>> brokerAddrTable = new ConcurrentHashMap<>();

    public MqClientApiImpl(NettyClientConfig nettyClientConfig) {
        this.nettyClientConfig = nettyClientConfig;
        this.remoteClient2 = new RemoteClient2(this.nettyClientConfig);
    }

    public void start() {
        this.remoteClient2.start();
    }

    public void shutdown() {
        this.remoteClient2.shutdown();
    }


    public FindBrokerResult selectBroker(String brokerName) {
        TreeMap<Integer, String> addrTables = this.brokerAddrTable.get(brokerName);
        if (addrTables == null) {
            return null;
        }
        String addr = addrTables.get(0);
        FindBrokerResult findBrokerResult = new FindBrokerResult();
        findBrokerResult.setAddr(addr);
        return findBrokerResult;
    }


    /**
     * 在broker端锁定队列
     *
     * @param messageQueue
     */
    public void lockQueue(MessageQueue messageQueue) {

    }

    /**
     * 在broker解锁队列
     *
     * @param messageQueue
     */
    public void unLockQueue(MessageQueue messageQueue, String group, String clientId) {
        MessageQueueAllLock messageQueueAllLock = new MessageQueueAllLock();
        messageQueueAllLock.setGroup(group).setClientId(clientId);
        Message message = new Message();
        message.setFlag((byte) 2);
        message.setType((byte) 16);
        message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_default);
        message.setByteBodys(RemotingSerializable.encode(messageQueueAllLock));
        message.setLength(message.getByteBodys().length);

        FindBrokerResult findBrokerResult = selectBroker(messageQueue.getBrokerName());
        if (findBrokerResult == null) {
            findBrokerResult = selectBroker(messageQueue.getBrokerName());
        }
        if (findBrokerResult == null) {
            return;
        }

        try {
            this.remoteClient2.invokeMsg2(message, findBrokerResult.getAddr());
        } catch (Exception e) {
            System.out.println("mq队列解锁异常：" + e.getMessage());
//            e.printStackTrace();
        }

    }

    public void sendHeartbeat(ClientInfo clientInfo) {
//        System.out.println("注册心跳数据");
        Message message = new Message();
        message.setFlag((byte) 2);
        message.setType((byte) 17);
        message.setNettyRequestWrapperCode((byte) ProcessorCode.hearbeat);
        message.setByteBodys(RemotingSerializable.encode(clientInfo));
        message.setLength(message.getByteBodys().length);
        Iterator<ConcurrentMap.Entry<String, TreeMap<Integer, String>>> iterator = this.brokerAddrTable.entrySet().iterator();
        while (iterator.hasNext()) {
            ConcurrentMap.Entry<String, TreeMap<Integer, String>> entry = iterator.next();
            for (String addr : entry.getValue().values()) {
                this.remoteClient2.invokeMsg2(message, addr);
            }
        }

    }

    public List<String> pullClientList(String group, String addr) {
        ClientInfo clientInfo = new ClientInfo();
        clientInfo.setGroup(group);
        Message message = new Message();
        message.setFlag((byte) 2);
        message.setType((byte) 18);
        message.setNettyRequestWrapperCode((byte) ProcessorCode.hearbeat);
        message.setByteBodys(RemotingSerializable.encode(clientInfo));
        message.setLength(message.getByteBodys().length);
        try {
            byte[] result = this.remoteClient2.invokeMsg(message, addr);
            if (result != null) {
                List<String> list = RemotingSerializable.decode(result, List.class);
                return list;
            }
        } catch (Exception e) {
            System.out.println("拉取消费列表异常：" + e.getMessage());
//            e.printStackTrace();
        }
        return null;
    }

    /**
     * 拉取消息
     *
     * @param pullRequest
     */
    public void pullMessage(PullRequest pullRequest, PullTask pullTask) {
        System.out.println(String.format("主题：%s,队列：%d,正在拉取消息，拉取位移：%d", pullRequest.getMessageQueue().getTopic(), pullRequest.getMessageQueue().getQueueId(), pullRequest.getNextOffset()));
        FindBrokerResult findBrokerResult = selectBroker(pullRequest.getMessageQueue().getBrokerName());
        if (findBrokerResult == null) {
            updateTopicRouteData(pullRequest.getMessageQueue().getTopic());
            findBrokerResult = selectBroker(pullRequest.getMessageQueue().getBrokerName());
        }
        if (findBrokerResult == null) {
            return;
        }
        ConsumeQueueData consumeQueueData = new ConsumeQueueData();
        consumeQueueData.setQueueId(pullRequest.getMessageQueue().getQueueId());
        consumeQueueData.setOffset(pullRequest.getNextOffset());
        consumeQueueData.setTopic(pullRequest.getMessageQueue().getTopic());
        consumeQueueData.setControllerStatus(ControllerStatus.PULL_MESSAGE);

        Message message = new Message();
        message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_msg);
        message.setFlag((byte) 2);
        message.setType((byte) 11);
        message.setByteBodys(RemotingSerializable.encode(consumeQueueData));
        message.setLength(message.getByteBodys().length);
        long t1 = System.currentTimeMillis();

        InvokeCallback invokeCallback = new InvokeCallback() {
            @Override
            public void callback(byte[] bytes) {
//                System.out.println(String.format("消息：%s，队列：%d，拉取消息完毕!", pullRequest.getMessageQueue().getTopic(), pullRequest.getMessageQueue().getQueueId()));

                long t2 = System.currentTimeMillis();
                if (bytes != null) {
                    PullResult pullResult = RemotingSerializable.decode(bytes, PullResult.class);
                    if (pullResult != null) {
                        switch (pullResult.getStatus()) {
                            case FOUND:
//                                System.out.println("有新消息，耗时：" + (t2 - t1) / 1000 + "秒");
                                pullTask.onSuccess(pullResult);
                                break;
                            default:
//                                System.out.println("没有消息，耗时：" + (t2 - t1) / 1000 + "秒");
                                PullResult error = new PullResult();
                                error.setStatus(PullResultStatus.NOT_FOUND);
                                pullTask.onSuccess(error);
                                break;
                        }
                    }
                } else {
//                    System.out.println("失联了" + (t2 - t1) / 1000 + "秒");
                    PullResult error = new PullResult();
                    error.setStatus(PullResultStatus.NOT_FOUND);
                    pullTask.onSuccess(error);
                }
            }
        };
        this.remoteClient2.invokeMsgAsync(message, findBrokerResult.getAddr(), invokeCallback);

    }

    /**
     * 拉取位移
     *
     * @param messageQueue
     * @return
     */
    public long getQueueOffset(MessageQueue messageQueue) {
        FindBrokerResult findBrokerResult = selectBroker(messageQueue.getBrokerName());
        if (findBrokerResult == null) {
            updateTopicRouteData(messageQueue.getTopic());
            findBrokerResult = selectBroker(messageQueue.getBrokerName());
        }
        if (findBrokerResult == null) {
            return 0l;
        }
        Message message = new Message();
        message.setFlag((byte) 1);
        message.setType((byte) 13);
        message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_offset);

        ConsumeQueueData consumeQueueData = new ConsumeQueueData();
        consumeQueueData.setTopic(messageQueue.getTopic());
        consumeQueueData.setQueueId(messageQueue.getQueueId());
        consumeQueueData.setOffset(0l);
        message.setByteBodys(RemotingSerializable.encode(consumeQueueData));
        message.setLength(message.getByteBodys().length);
        try {
            byte[] bytes = this.remoteClient2.invokeMsg(message, findBrokerResult.getAddr());
            if (bytes != null) {
                ConsumeQueueData consumeQueueData1 = RemotingSerializable.decode(bytes, ConsumeQueueData.class);

                return consumeQueueData1.getOffset();
            }
        } catch (Exception e) {
//            e.printStackTrace();
            System.out.println("拉取消费位移异常：" + e.getMessage());
        }
        return 0l;
    }

    //持久化消费位移
    public void persitQueueOffset(MessageQueue messageQueue, Long offset) {
        FindBrokerResult findBrokerResult = selectBroker(messageQueue.getBrokerName());
        if (findBrokerResult == null) {
            updateTopicRouteData(messageQueue.getTopic());
            findBrokerResult = selectBroker(messageQueue.getBrokerName());
        }
        if (findBrokerResult == null) {
            return;
        }

        Message message = new Message();
        message.setFlag((byte) 1);
        message.setType((byte) 12);
        message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_offset);

        ConsumeQueueData consumeQueueData = new ConsumeQueueData();
        consumeQueueData.setTopic(messageQueue.getTopic());
        consumeQueueData.setQueueId(messageQueue.getQueueId());
        consumeQueueData.setOffset(offset);
        message.setByteBodys(RemotingSerializable.encode(consumeQueueData));
        message.setLength(message.getByteBodys().length);
        try {
            this.remoteClient2.invokeMsg2(message, findBrokerResult.getAddr());
        } catch (Exception e) {
            System.out.println("持久化消费位移异常：" + e.getMessage());
//            e.printStackTrace();
        }

    }

    //从服务器更新topicRouteData
    public byte[] updateTopicRouteData(String topic) {
        if (StringUtils.isEmpty(topic)) {
            throw new MyException("topic没有订阅！");
        }
        try {
            Message message = new Message();
            message.setFlag((byte) 1);
            message.setType((byte) 15);
            message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_default);
            message.setByteBodys(topic.getBytes(Charset.forName("utf-8")));
            message.setLength(message.getByteBodys().length);
            byte[] bytes = this.remoteClient2.invokeMsg(message, GlobalConfigUtil.getNameSrv());
            return bytes;
        } catch (Exception e) {
            System.out.println("更新topic路由信息异常：" + e.getMessage());
            e.printStackTrace();
        }
        return null;
    }

    public Set<MessageQueue> lockAllQueue(MessageQueueAllLock messageQueueAllLock, String brokerName) {
        FindBrokerResult findBrokerResult = selectBroker(brokerName);
        if (findBrokerResult == null) {
            findBrokerResult = selectBroker(brokerName);
        }
        if (findBrokerResult == null) {
            throw new MyException("连接broker失败！");
        }
        Message message = new Message();
        message.setType((byte) 15);
        message.setFlag((byte) 2);
        message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_offset);
        message.setByteBodys(RemotingSerializable.encode(messageQueueAllLock));
        message.setLength(message.getByteBodys().length);
        try {
            byte[] result = this.remoteClient2.invokeMsg(message, findBrokerResult.getAddr());
            if (result != null) {
                MessageQueueAllLock allLock = RemotingSerializable.decode(result, MessageQueueAllLock.class);
                if (allLock != null) {
                    return allLock.getMessageQueueSet();
                }
            }
        } catch (Exception e) {
//            e.printStackTrace();
            System.out.println("锁定队列异常：" + e.getMessage());
        }
        return null;
    }

    public ConcurrentMap<String, TreeMap<Integer, String>> getBrokerAddrTable() {
        return brokerAddrTable;
    }
}
